package cn.jly.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 有状态转换操作
 *
 * @author lanyangji
 * @date 2019/12/5 19:10
 */
object SparkStreaming06_UpdateState {

  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming06_UpdateState")
    val streamingContext: StreamingContext = new StreamingContext(sparkConf, Seconds(5))
    // 设置检查点目录
    streamingContext.checkpoint("check-point")

    // 监控并读取端口的数据写入
    val socketDStream: ReceiverInputDStream[String] = streamingContext.socketTextStream("hadoop102", 9999)

    // 处理，做wordCount
    // 无状态转换
    //socketDStream.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_).print()

    // 有状态转换
    socketDStream
      .flatMap(_.split(" "))
      .map((_, 1))
      .updateStateByKey[Int] {
        // values为当前批次单词的频度集合（1的集合），state为以往批次单词频度之和
        (values: Seq[Int], state: Option[Int]) => {

          val currentCount: Int = values.sum
          val previousCount: Int = state.getOrElse(0)
          Some(currentCount + previousCount)
        }
      }
      .print()

    // 启动
    streamingContext.start()
    streamingContext.awaitTermination()
  }
}
